package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo15StudentTest1 {
  def main(args: Array[String]): Unit = {
    //求年级总分前10的学生各科分数的详细信息
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("MapValues算子演示")

    val sc: SparkContext = new SparkContext(conf)

    val linesRDD: RDD[String] = sc.textFile("spark/data/score.txt")

    val idWithScoreRDD: RDD[(String, String, Int)] = linesRDD.map((line: String) => {
      line.split(",") match {
        case Array(id: String, subject_id: String, score: String) =>
          (id, subject_id, score.toInt)
      }
    })


    val array1: Array[String] = idWithScoreRDD
      .map((t3: (String, String, Int)) => (t3._1, t3._3))
      .reduceByKey(_ + _)
      .sortBy((kv: (String, Int)) => -kv._2)
      .take(10)
      .map(_._1)

    idWithScoreRDD.filter((t3: (String, String, Int)) => {
      val bool: Boolean = array1.contains(t3._1)
      if(bool){
        println("存在")
      }
      bool
    }).foreach((t3: (String, String, Int)) => {
      println("==========================")
      println(t3)
    })


  }
}
